package com.atguigu.flink.chapter11;
import com.atguigu.flink.bean.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.*;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

import static org.apache.flink.table.api.Expressions.*;

public class Flink14_Window_Over {
       public static void main(String[] args) {
               Configuration configuration = new Configuration();
               //web  UI端口
               configuration.setInteger("rest.prot",10000);
               StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);
               env.setParallelism(1);

               //因为加了window 后变成普通的流的，不是source？？？！！！ 不是 DataStreamSource 了
           DataStream<WaterSensor> waterSensorStream =
                   env.fromElements(
                           new WaterSensor("sensor_1", 1000L, 10),
                           new WaterSensor("sensor_1", 5000L, 20),
                           new WaterSensor("sensor_1", 4000L, 40),
                           new WaterSensor("sensor_1", 4000L, 50),
                           new WaterSensor("sensor_2", 6000L, 30),
                           new WaterSensor("sensor_2", 7000L, 60))
                   .assignTimestampsAndWatermarks(
                           WatermarkStrategy
                                    .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                   //事件事件来自于时间戳
                                   .withTimestampAssigner((event, timestamp) -> event.getTs())
                   );


           StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //  每隔5s 统计最近5s 内 每个传感器的水位和
           Table table = tableEnv.fromDataStream(waterSensorStream, $("id"), $("ts").rowtime(), $("vc"));
           // sum(vc ) over(partition by id order by ts rows between unbound  row and current row)
           // ts 在flink 中  只能升序，不能降序，只有topN的时候，才能使用 ，因为流只能往前跑，不能倒退

           // 必须加别名，不然类型错误，采用父类
           OverWindow w = Over                                            //  默认的preceding(Expressions.UNBOUNDED_ROW).following(Expressions.CURRENT_ROW)
                   .partitionBy($("id")).orderBy($("ts")).preceding(Expressions.UNBOUNDED_ROW).following(Expressions.CURRENT_ROW).as("w");


           //                                                                    多尝试几次，得出他们的特性    往前一行 即2行相加
           OverWindow w2 = Over.partitionBy($("id")).orderBy($("ts")).preceding(rowInterval(1L)).following(CURRENT_ROW).as("w");

           // range  根据时间来的 可以选择不同的正交性 同时到的数据  进入同一个窗口，而ROW 是分别进入为2个窗口
           OverWindow w3 = Over.partitionBy($("id")).orderBy($("ts")).preceding(UNBOUNDED_RANGE).following(CURRENT_RANGE).as("w");
          // range  根据时间来的                                             往前1s 的事件时间 累加现在的事件数据
           OverWindow w4 = Over.partitionBy($("id")).orderBy($("ts")).preceding(lit(1).second()).following(CURRENT_RANGE).as("w");

           // 只要是字段 必须是 $("w") 填入 不能"w" 会报错
           table
                   .window(w4)
                   .select($("id"),$("ts"),$("vc"),$("vc").sum().over($("w")).as("vc_sum"))
                   .execute()
                   .print();

           //  flink 系统里面   还有正交性  当数据的ts 时间  一样的，
           //  可以选择不同的正交性，从而进行划分到不同的窗口！！！ ROW 与 RANGE 的不同




       }
}
